package net.oschina.simpleplus.erpc.consumer;

import java.util.concurrent.ThreadPoolExecutor;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import net.oschina.simpleplus.erpc.Launcher;
import net.oschina.simpleplus.erpc.api.ERpcContainer;
import net.oschina.simpleplus.erpc.support.ERpcMessageEncoder;

public class ERpcServer implements Launcher{
	
	private EventLoopGroup boss = new NioEventLoopGroup();
	private EventLoopGroup worker = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);
	
	private ThreadPoolExecutor buinessExecutor;
	private ERpcContainer eRpcContainer;

	public ERpcServer(ThreadPoolExecutor executor, ERpcContainer eRpcContainer) {
		super();
		this.buinessExecutor = executor;
		this.eRpcContainer = eRpcContainer;
	}

	@Override
	public void start(int port)throws Exception{
		try{
		ServerBootstrap b =
				new ServerBootstrap()
				.group(boss, worker)
				.channel(NioServerSocketChannel.class)
				.option(ChannelOption.SO_BACKLOG, 100)
	             .handler(new LoggingHandler(LogLevel.INFO))
	             .childHandler(new ChannelInitializer<SocketChannel>() {
	                 @Override
	                 public void initChannel(SocketChannel ch) throws Exception {
	                     ch.pipeline().addLast(
	                    		 new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4),
	                    		 new ERpcDecoder(),
	                    		 new ERpcMessageEncoder(),
	                    		 new ERpcHandler(buinessExecutor, eRpcContainer));
	                 }
	             });
	 
	            // Start the server.
	            ChannelFuture f = b.bind(port).sync();
	 
	            // Wait until the server socket is closed.
	            f.channel().closeFuture().sync();
	        } finally {
	        	shutdown();
	        }
	}
	
	@Override
	public void shutdown(){
		boss.shutdownGracefully();
        worker.shutdownGracefully();
        
        buinessExecutor.shutdownNow();
	}
	
	
}
